package com.huawei.dli.flink.sye.avro;

import java.io.IOException;

import com.huawei.hwclouds.drs.avro.Record;
import org.apache.avro.message.BinaryMessageDecoder;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.formats.avro.typeutils.AvroTypeInfo;

/**
 * @author joker
 */
public class AvroDerializationSchema implements DeserializationSchema<Record> {
  private transient BinaryMessageDecoder<Record> decoder;

  @Override
  public Record deserialize(byte[] message) throws IOException {
    if (decoder == null) {
      decoder = Record.getDecoder();
      decoder.addSchema(Record.getClassSchema());
    }
    return decoder.decode(message);
  }

  @Override
  public boolean isEndOfStream(Record nextElement) {
    return false;
  }

  @Override
  @SuppressWarnings("unchecked")
  public TypeInformation<Record> getProducedType() {
    return new AvroTypeInfo(Record.class);
  }
}
